package cn.tedu.sql

import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.table.api.Table
import org.apache.flink.table.api.scala.BatchTableEnvironment
import org.apache.flink.types.Row

/**
 * @author Amos
 * @date 2022/5/24
 */

object BatchSqlDemo {
  def main(args: Array[String]): Unit = {
    // 1. 批处理的环境
    val env = ExecutionEnvironment.getExecutionEnvironment

    // 2. 构建table的环境
    val tableEnv: BatchTableEnvironment = BatchTableEnvironment.create(env)

    // 3. 构建数据源
    import org.apache.flink.api.scala._
    val source: DataSet[Order] = env.fromElements(
      Order(1, "zhangsan", "2018-10-20 15:30", 358.5),
      Order(2, "zhangsan", "2018-10-20 16:30", 131.5),
      Order(3, "lisi", "2018-10-20 16:30", 127.5),
      Order(4, "lisi", "2018-10-20 16:30", 328.5),
      Order(5, "lisi", "2018-10-20 16:30", 432.5),
      Order(6, "zhaoliu", "2018-10-20 22:30", 451.0),
      Order(7, "zhaoliu", "2018-10-20 22:30", 362.0),
      Order(8, "zhaoliu", "2018-10-20 22:30", 364.0),
      Order(9, "zhaoliu", "2018-10-20 22:30", 341.0)
    )
    /*
    zhaoliu,1518.0,451.0,341.0,4
    lisi,888.5,432.5,127.5,3
    zhangsan,490.0,358.5,131.5,2
     */


    // 4. 数据的处理 sql
    // 将数据集注册成一张表
    tableEnv.createTemporaryView("t_order", source)

    // 使用Flink SQL 统计用户消费订单的总金额、最大金额、最小金额、订单总数
    val sql =
      """
        |select
        |name,
        |sum(money) totalMoney,
        |max(money) mxaMoney,
        |min(money) minMoney,
        |count(1) totalCount
        |from t_order
        |group by name
        |""".stripMargin

    // 执行sql
    val table: Table = tableEnv.sqlQuery(sql)

    val result: DataSet[Row] = tableEnv.toDataSet[Row](table)

    // 5. 结果的sink
    result.print()
//    env.execute() // 批处理不需要写这句话
  }

}

case class Order(id: Int, name: String, createTime: String, money: Double)
